package cn.yj.core.task.delay;

import cn.yj.commons.utils.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.xml.rpc.ServiceException;
import java.util.Set;
import java.util.concurrent.*;

/**
 * 延迟队列处理
 *
 * @author 永健
 * @since 2021-05-12 13:49
 */
public abstract class AbstractDelayTaskHandler<T extends BaseDelayTaskEntity> {
    private static final Logger log = LoggerFactory.getLogger(AbstractDelayTaskHandler.class);
    /**
     * 并发安全的set
     * ConcurrentHashMap<K,Boolean> 对象
     */
    private static final Set<String> UNIQUE = ConcurrentHashMap.newKeySet();

    private static final int corePoolSize = 4;

    private static ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(corePoolSize,
            new BasicThreadFactory.Builder().namingPattern("schedule" + "-pool-%d").daemon(true).build(),
            new ThreadRejectPolicy()) {
        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            super.afterExecute(r, t);
        }
    };

    private static class ThreadRejectPolicy implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            log.warn("schedule delay 线程池关闭");
            log.warn("queue size={}", executor.getQueue().size());
            log.warn("executor queueName={}", executor.getQueue().getClass().getName());
            log.warn("executor poolSize={}", executor.getPoolSize());
            log.warn("executor activeCount={}", executor.getActiveCount());
            log.warn("executor taskCount={}", executor.getTaskCount());
            log.warn("executor largestPoolSize={}", executor.getLargestPoolSize());
            log.warn("");
        }
    }

    public void execute(T taskEntity) {
        //if (!UNIQUE.contains(taskEntity.getUnique())) {
            scheduledExecutorService.schedule(taskEntity.getTask(), taskEntity.getDelay(), taskEntity.getTimeUnit());
        //}
    }

    protected CallBackFinish<T> newCallBack() {
        return (e, t) -> {
            if (StringUtils.isNull(e) || (e instanceof ServiceException)) {
                UNIQUE.remove(t.getUnique());
                return;
            }
            execute(t);
        };
    }

    /**
     * 初始化延迟任务
     */
    public abstract void initTask();
}
